#include "vl_dispatcher.h"

using namespace VL;

dispatch_base_msg::~dispatch_base_msg() {}

dispatch_runnable_msg::dispatch_runnable_msg(dispatch_run_cb cb): _cb(cb) {}
void dispatch_runnable_msg::invoke() {
	if(this->_cb) {
		this->_cb();
	}
}
void dispatch_runnable_msg::cancel() {

}

dispatcher::dispatcher(const std::string& name): 
	status(dispatcher_status::Init), name(name) { }
dispatcher::~dispatcher() {
}
void dispatcher::start() {
	this->mutex.lock();
	if(this->status != dispatcher_status::Init) {
		this->mutex.unlock();
		return;
	}
	this->status = dispatcher_status::Started;
	this->tid = std::this_thread::get_id();
	this->mutex.unlock();
	this->loop();
}
void dispatcher::start_sync() {
	this->start();
}
bool dispatcher::is_async() {
	return this->tid != std::this_thread::get_id();
}

void dispatcher::dispatch_run(const dispatch_run_cb& cb, bool runWhenSameThread) {
	if(this->status > dispatcher_status::Started) {
		return;
	}
	this->dispatch_msg(new dispatch_runnable_msg(cb), runWhenSameThread);
}
void dispatcher::dispatch_first_run(const dispatch_run_cb& cb, bool runWhenSameThread) {
	if(this->status > dispatcher_status::Started) {
		return;
	}
	this->dispatch_msg_at_first(new dispatch_runnable_msg(cb), runWhenSameThread);
}

void dispatcher::loop() {
	dispatch_base_msg* msg = nullptr;
	while(this->status <= dispatcher_status::SafeQuiting) {
		{
			std::unique_lock<std::mutex> locker(this->mutex);
			if(this->queue.empty()) {
				if(this->status == dispatcher_status::SafeQuiting) {
					break;
				}
				this->cond.wait_for(locker, std::chrono::milliseconds(1000));
			}
			if(this->status > dispatcher_status::SafeQuiting) {
				break;
			}
			if(this->queue.empty()) {
				if(this->status == dispatcher_status::SafeQuiting) {
					break;
				}
				continue;
			}
			msg = this->queue.front();
			this->queue.pop_front();
		}
		if(msg != nullptr) {
			msg->invoke();
			delete msg;
		}
	}
	// 结束循环, 回收全部未消费的消息
	while(!this->queue.empty()) {
		msg = this->queue.front();
		this->queue.pop_front();
		msg->cancel();
		delete msg;
	}
}

void dispatcher::dispatch_msg(dispatch_base_msg* msg, bool runWhenSameThread) {
	if(msg == nullptr) {
		return;
	}
	this->mutex.lock();
	if(this->status > dispatcher_status::Started) {
		this->mutex.unlock();
		msg->cancel();
		delete msg;
		return;
	}
	else if(this->status < dispatcher_status::Started) {
		this->queue.push_back(msg);
		this->mutex.unlock();
		return;
	} else if(!runWhenSameThread || this->tid != std::this_thread::get_id()) {
		this->queue.push_back(msg);
		this->mutex.unlock();
		this->cond.notify_all();
		return;
	}
	this->mutex.unlock();
	msg->invoke();
	delete msg;
}
void dispatcher::dispatch_msg_at_first(dispatch_base_msg* msg, bool runWhenSameThread) {
	if(msg == nullptr) {
		return;
	}
	this->mutex.lock();
	if(this->status > dispatcher_status::Started) {
		this->mutex.unlock();
		msg->cancel();
		delete msg;
		return;
	}
	else if(this->status < dispatcher_status::Started) {
		this->queue.push_front(msg);
		this->mutex.unlock();
		return;
	} else if(!runWhenSameThread || this->tid != std::this_thread::get_id()) {
		this->queue.push_front(msg);
		this->mutex.unlock();
		this->cond.notify_all();
		return;
	}
	this->mutex.unlock();
	msg->invoke();
	delete msg;
}
void dispatcher::quit() {
	this->mutex.lock();
	if(this->status > dispatcher_status::SafeQuiting) {
		this->mutex.unlock();
		return;
	}
	this->status = dispatcher_status::Quiting;
	this->mutex.unlock();
	this->cond.notify_all();
}
void dispatcher::quit_sync() {
	this->quit();
}
void dispatcher::quit_safety() {
	this->mutex.lock();
	if(this->status > dispatcher_status::Started) {
		this->mutex.unlock();
		return;
	}
	this->status = dispatcher_status::SafeQuiting;
	this->mutex.unlock();
	this->cond.notify_all();
}
void dispatcher::quit_safety_sync() {
	this->quit_safety();
}


async_dispatcher::async_dispatcher(const std::string& name): dispatcher(name), threadPtr(nullptr) {}
async_dispatcher::~async_dispatcher() {
	this->quit_sync();
}

void async_dispatcher::start() {
	this->mutex.lock();
	if(this->status != dispatcher_status::Init) {
		this->mutex.unlock();
		return;
	}
	this->status = dispatcher_status::Starting;
	this->threadPtr = new std::thread([&]() {
		this->mutex.lock();
		if(this->status == dispatcher_status::Starting) {
			this->status = dispatcher_status::Started;
			this->tid = std::this_thread::get_id();
			this->statusCond.notify_all();
		}
		this->mutex.unlock();
		this->loop();
	});
	this->mutex.unlock();
}
void async_dispatcher::start_sync() {
	this->start();
	std::unique_lock<std::mutex> locker(this->mutex);
	if(this->status == dispatcher_status::Starting) {
		this->statusCond.wait(locker);
	}
}

void async_dispatcher::quit() {
	this->mutex.lock();
	if(this->status > dispatcher_status::SafeQuiting) {
		this->mutex.unlock();
		return;
	} else if(this->status == dispatcher_status::Starting) {
		this->statusCond.notify_all();
	} else if(this->status == dispatcher_status::Started) {
		this->cond.notify_all();
	}
	if(this->threadPtr == nullptr) {
		if(this->status == dispatcher_status::SafeQuiting) {
			this->status = dispatcher_status::Quiting;
		} else {
			this->status = dispatcher_status::Quit;
		}
	} else {
		this->status = dispatcher_status::Quiting;
	}
	this->mutex.unlock();
}
void async_dispatcher::quit_sync() {
	this->mutex.lock();
	std::thread* _threadPtr = nullptr;
	switch(this->status) {
		case dispatcher_status::Quit:
			this->mutex.unlock();
			return;
		case dispatcher_status::SafeQuiting:
			this->status = dispatcher_status::Quiting;
		case dispatcher_status::Quiting:
			if(this->threadPtr != nullptr) {
				_threadPtr = this->threadPtr;
				this->threadPtr = nullptr;
			}
			break;
		case dispatcher_status::Starting:
			this->statusCond.notify_all();
		default:
			if(this->threadPtr == nullptr) {
				this->status = dispatcher_status::Quit;
				this->mutex.unlock();
				return;
			}
			_threadPtr = this->threadPtr;
			this->threadPtr = nullptr;
			this->status = dispatcher_status::Quiting;
			break;
	}
	this->mutex.unlock();
	this->cond.notify_all();

	if(_threadPtr != nullptr && _threadPtr->joinable()) {
		_threadPtr->join();
	}

	std::unique_lock<std::mutex> locker(this->mutex);
	if(this->status == dispatcher_status::Quit) {
		return;
	}
	if(_threadPtr == nullptr) {
		this->statusCond.wait(locker);
	} else {
		this->status = dispatcher_status::Quit;
		this->statusCond.notify_all();
	}
}
void async_dispatcher::quit_safety() {
	this->mutex.lock();
	if(this->status > dispatcher_status::Started) {
		this->mutex.unlock();
		return;
	} else if(this->status == dispatcher_status::Starting) {
		this->statusCond.notify_all();
	} else if(this->status == dispatcher_status::Started) {
		this->cond.notify_all();
	}
	this->status = this->threadPtr == nullptr ? dispatcher_status::Quit : dispatcher_status::SafeQuiting;
	this->mutex.unlock();
}
void async_dispatcher::quit_safety_sync() {
	this->mutex.lock();
	std::thread* _threadPtr = nullptr;
	switch(this->status) {
		case dispatcher_status::Quit:
			this->mutex.unlock();
			return;
		case dispatcher_status::SafeQuiting:
		case dispatcher_status::Quiting:
			if(this->threadPtr != nullptr) {
				_threadPtr = this->threadPtr;
				this->threadPtr = nullptr;
			}
			break;
		case dispatcher_status::Starting:
			this->statusCond.notify_all();
		default:
			if(this->threadPtr == nullptr) {
				this->status = dispatcher_status::Quit;
				this->mutex.unlock();
				return;
			}
			_threadPtr = this->threadPtr;
			this->threadPtr = nullptr;
			this->status = dispatcher_status::SafeQuiting;
			break;
	}
	this->mutex.unlock();
	this->cond.notify_all();

	if(_threadPtr != nullptr && _threadPtr->joinable()) {
		_threadPtr->join();
	}

	std::unique_lock<std::mutex> locker(this->mutex);
	if(this->status == dispatcher_status::Quit) {
		return;
	}
	if(_threadPtr == nullptr) {
		this->statusCond.wait(locker);
	} else {
		this->status = dispatcher_status::Quit;
		this->statusCond.notify_all();
	}
}